package cn.com.itg.function.process;

import cn.com.itg.pojo.domain.entity.MATDOC;
import cn.com.itg.pojo.domain.entity.ZTTCIDENTICAL;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.io.Serializable;
import java.math.BigDecimal;
import java.util.List;
import java.util.Objects;

public class MatdocZqrwerksBPFunction extends KeyedBroadcastProcessFunction<String, MATDOC, List<ZTTCIDENTICAL>, MATDOC> implements Serializable {

    private final String zttTag = "zttInfo";
    // 配置信息未加载完毕时先缓存数据流元素
    private transient ListState<MATDOC> matdocList;

    // 未设置定时输出缓存流元素的定时器，默认未设置
    private transient ValueState<Boolean> setTimer;

    @Override
    public void open(Configuration parameters) throws Exception {
        //临时存放未处理的matdoc的ListState
        matdocList = getRuntimeContext().getListState(new ListStateDescriptor<>("matdoc", TypeInformation.of(new TypeHint<MATDOC>() {
        })));
        //控制定时器
        setTimer = getRuntimeContext().getState(new ValueStateDescriptor<>("time_sets", TypeInformation.of(new TypeHint<Boolean>() {
        })));

    }

    @Override
    public void processElement(MATDOC matdoc, ReadOnlyContext readOnlyContext, Collector<MATDOC> collector) throws Exception {
        List<ZTTCIDENTICAL> ztt_zqrwerks = readOnlyContext.getBroadcastState(
                new MapStateDescriptor<>(
                        "ztt_zqrwerks",
                        TypeInformation.of(new TypeHint<String>() {
                        }),
                        TypeInformation.of(new TypeHint<List<ZTTCIDENTICAL>>() {
                        })))
                .get(zttTag);
        //如果获取到ZTTCIDENTICAL列表信息不为空且数量大于0，则判断是否有缓存未处理信息
        if (Objects.nonNull(ztt_zqrwerks) && ztt_zqrwerks.size() > 0) {
            //如果有缓存
            if (null != setTimer.value() && setTimer.value()) {
                matdocList.get().forEach(matdocInfo -> {
                    output(matdoc, ztt_zqrwerks, collector);
                });
                // 处理后把缓存清空
                matdocList.clear();
                // 同时修改定时处理缓存的标志，说明缓存已经没数据了，不需要定时处理缓存数据了
                setTimer.update(false);
            }
            // 这里是正常处理流元素
            output(matdoc, ztt_zqrwerks, collector);
        } else {
            // 如果ZTTCIDENTICAL列表信息还没有加载完毕，先把matdoc流元素缓存起来
            matdocList.add(matdoc);
            // 只需要一个定时器就行了，到时执行一次就可以把缓存的所有数据处理掉
            if (null == setTimer.value() || !setTimer.value()) {
                // 注册定时器，如果后面没来数据，默认10秒后把缓存的数据发送到下游
                long fireTime = readOnlyContext.timerService().currentProcessingTime() + 10000;
                readOnlyContext.timerService().registerProcessingTimeTimer(fireTime);
                setTimer.update(true);
            }
        }

    }

    @Override
    public void processBroadcastElement(List<ZTTCIDENTICAL> zttcidenticals, Context context, Collector<MATDOC> collector) throws Exception {
        BroadcastState<String, List<ZTTCIDENTICAL>> ztt_state = context.getBroadcastState(
                new MapStateDescriptor<>(
                        "ztt_zqrwerks",
                        TypeInformation.of(new TypeHint<String>() {
                        }),
                        TypeInformation.of(new TypeHint<List<ZTTCIDENTICAL>>() {
                        })));
        ztt_state.put(zttTag, zttcidenticals);
    }

    /**
     * 定时输出缓存的数据
     * 配置流未加载完毕时缓存此时过来的数据流元素，缓存的数据流元素只有等下一个流元素来的时候才会触发后续的处理操作，这里的定时器可以定时输出缓存的数据
     *
     * @param timestamp 定时器触发时间
     * @param ctx       上下文
     * @param out       输出器
     */
    @Override
    public void onTimer(long timestamp, KeyedBroadcastProcessFunction<String, MATDOC, List<ZTTCIDENTICAL>, MATDOC> .OnTimerContext ctx, Collector<MATDOC> out) throws Exception {
        List<ZTTCIDENTICAL> ztt_zqrwerks = ctx.getBroadcastState(new MapStateDescriptor<>(
                "ztt_zqrwerks",
                TypeInformation.of(new TypeHint<String>() {
                }),
                TypeInformation.of(new TypeHint<List<ZTTCIDENTICAL>>() {
                })))
                .get(zttTag);
        matdocList.get().forEach(matdoc -> {
            output(matdoc,ztt_zqrwerks,out);
        });
        // 输出后把缓存清空防止重复输出
        matdocList.clear();
        // 执行完任务删除定时器
        ctx.timerService().deleteProcessingTimeTimer(timestamp);
        // 同时更新定时器的状态
        setTimer.update(false);
    }

    /**
     * @param matdoc
     * @param zttcidenticals
     * @param collector
     * @return
     * @throws
     * @author chenzh
     * @description 输出到下个算子
     * @date 2022-10-07 12:10:20
     **/
    private void output(MATDOC matdoc, List<ZTTCIDENTICAL> zttcidenticals, Collector<MATDOC> collector) {
        if (Objects.isNull(zttcidenticals)) {
            return;
        }
        zttcidenticals.stream()
                .forEach(zttcidentical -> {
                    //1.WERKS==ZDTYPE
                    if (!Objects.equals(matdoc.getWerks(), zttcidentical.getZdtype())) {
                        return;
                    }
                    //2.BUDAT>=ZFIELD2
                    BigDecimal budat = new BigDecimal(matdoc.getBudat());
                    BigDecimal zfield2 = new BigDecimal(zttcidentical.getZfield2());
                    if (budat.compareTo(zfield2) > -1) {
                        collector.collect(matdoc);
                    }
                });
    }


}
